package com.study.source;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * 自定义source
 * 功能：对所有输入数据传一个前缀和后缀
 */
public class MySource extends AbstractSource implements Configurable , PollableSource {

    private String prefix;
    private String subfix;
    private Long delay;

    /**
     * 读取配置文件
     * 声明前后缀，前缀有默认值，后缀没有
     * @param context
     */
    @Override
    public void configure(Context context) {
        //在配置文件中的形式 例子：a1.sources.r1.prefix = prefix-
        prefix = context.getString("prefix", "prefix-");
        subfix = context.getString("subfix");
        delay = context.getLong("delay",2000L);
    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status;
        //生命事件
        Event event = new SimpleEvent();
        //头信息集合
        HashMap<String,String> header = new HashMap<>();
        ArrayList<Event> events = new ArrayList<>();
        //按住 ctrl+alt+t可以选择包裹
        try {
            event.setHeaders(header);
            for (int i = 0; i < 5; i++) {
                event.setBody((prefix+"huangmiaochun\t"+i+subfix).getBytes(StandardCharsets.UTF_8));
                events.add(event);
            }
            Thread.sleep(delay);
            getChannelProcessor().processEventBatch(events);
            status = Status.READY;
        } catch (Exception e) {
            status = Status.BACKOFF;
            e.printStackTrace();
        }
        return status;
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }
}
